Reviewing files that changed from the base of the PR and between 10ebd3a and e5213f7.

• include/fluent-bit/aws/flb_aws_msk_iam.h (0 hunks)
• plugins/in_kafka/in_kafka.c (2 hunks)
• plugins/in_kafka/in_kafka.h (0 hunks)
• plugins/out_kafka/kafka.c (0 hunks)
• plugins/out_kafka/kafka_config.c (3 hunks)
• plugins/out_kafka/kafka_config.h (0 hunks)
• src/aws/flb_aws_credentials_ec2.c (1 hunks)
• src/aws/flb_aws_credentials_profile.c (1 hunks)
• src/aws/flb_aws_credentials_sts.c (2 hunks)
• src/aws/flb_aws_msk_iam.c (11 hunks)

• include/fluent-bit/aws/flb_aws_msk_iam.h
• plugins/out_kafka/kafka.c
• plugins/out_kafka/kafka_config.h
• plugins/in_kafka/in_kafka.h

src/aws/flb_aws_credentials_sts.c (1)

478-489:

src/aws/flb_aws_credentials_profile.c (1)

664-673: LGTM - Appropriate log level adjustment for missing credentials file.

Downgrading ENOENT from a conditional error/debug to always debug is correct since a missing shared credentials file is a normal condition when using other authentication methods (environment variables, instance profiles, etc.).

src/aws/flb_aws_credentials_ec2.c (1)

132-133: No concerns - formatting change only.

src/aws/flb_aws_msk_iam.c (5)

41-53: LGTM - Token lifetime and struct changes are appropriate.

The 5-minute token lifetime aligns with AWS SDK conventions, and the struct changes properly reflect the new broker-derived region approach.

---

135-188: Region extraction logic looks correct.

The function properly handles both MSK Standard (*.kafka.<region>.amazonaws.com) and MSK Serverless (*.kafka-serverless.<region>.amazonaws.com) broker formats. The backwards scan from .amazonaws.com correctly locates the region segment.

---

190-488: Payload generation logic is correctly implemented.

The SigV4 signing, canonical request construction, and base64 URL encoding follow AWS specifications. Resource cleanup in both success and error paths is thorough.

---

490-587: OAuth callback implementation is well-structured.

The callback properly:

1. Refreshes credentials before retrieval (lines 532-536)
2. Destroys credentials after use (lines 550, 573)
3. Handles all error paths with appropriate rd_kafka_oauthbearer_set_token_failure calls
4. Uses a fixed 5-minute lifetime which librdkafka will refresh at ~80% (4 minutes)

---

718-737: LGTM - Proper resource cleanup.

The destroy function correctly cleans up all allocated resources (provider, TLS, region) with appropriate null checks.

plugins/in_kafka/in_kafka.c (2)

271-294: LGTM - SASL mechanism conversion logic is correct.

The automatic conversion from aws_msk_iam to OAUTHBEARER and defaulting security.protocol to SASL_SSL provides the expected simplified configuration experience.

---

336-342: LGTM - Opaque context correctly includes Kafka context.

Passing &ctx->kafka to flb_kafka_opaque_set enables the MSK IAM module to access broker configuration for region extraction.

plugins/out_kafka/kafka_config.c (2)

61-84: LGTM - Consistent SASL mechanism handling with in_kafka.

The conversion logic mirrors the input plugin implementation, ensuring consistent behavior across both Kafka plugins.

---

251-270: LGTM - SASL background callbacks handling is appropriate.

Enabling background callbacks ensures OAuth tokens refresh even on idle connections. Making failure non-fatal is reasonable since token refresh will still work through rd_kafka_poll, though with less reliability on idle connections.

Reviewing files that changed from the base of the PR and between a6d39b1 and 491287a.

• include/fluent-bit/aws/flb_aws_msk_iam.h (1 hunks)
• plugins/in_kafka/in_kafka.c (4 hunks)
• plugins/in_kafka/in_kafka.h (1 hunks)
• plugins/out_kafka/kafka.c (0 hunks)
• plugins/out_kafka/kafka_config.c (4 hunks)
• plugins/out_kafka/kafka_config.h (1 hunks)
• src/aws/flb_aws_credentials_ec2.c (1 hunks)
• src/aws/flb_aws_credentials_profile.c (1 hunks)
• src/aws/flb_aws_credentials_sts.c (2 hunks)
• src/aws/flb_aws_msk_iam.c (12 hunks)
• src/flb_kafka.c (1 hunks)

• plugins/out_kafka/kafka.c

• plugins/out_kafka/kafka_config.h
• include/fluent-bit/aws/flb_aws_msk_iam.h
• plugins/in_kafka/in_kafka.h

src/aws/flb_aws_credentials_sts.c (1)

173-186: Formatting-only change; behavior unchanged

The added blank lines after the debug logs in refresh_fn_sts and refresh_fn_eks are cosmetic and do not affect control flow or error handling. No further changes needed.

Also applies to: 478-489

src/flb_kafka.c (1)

96-100: Correctly destroy rd_kafka_conf_t on error

Switching from flb_free(kafka_cfg) to rd_kafka_conf_destroy(kafka_cfg) in the error path aligns with how rd_kafka_conf_t is created and should be freed. This avoids undefined behavior and potential leaks when rd_kafka_conf_new() succeeds but flb_kafka_conf_create() fails later.

Please double-check against the librdkafka rd_kafka_conf_new / rd_kafka_conf_destroy docs that rd_kafka_conf_destroy() is the expected way to free this object in your target version.

src/aws/flb_aws_credentials_ec2.c (1)

128-139: Whitespace-only change

The extra blank line after the EC2 IMDS provider refresh debug log is stylistic only and has no runtime impact.

src/aws/flb_aws_credentials_profile.c (1)

646-674: More appropriate log level for missing shared credentials file

When the shared credentials file is absent (errno == ENOENT), logging via AWS_CREDS_DEBUG instead of AWS_CREDS_ERROR_OR_DEBUG avoids spurious error-level noise while still returning an error to the caller. This matches how the shared config path already treats ENOENT and looks good.

plugins/in_kafka/in_kafka.c (3)

347-357: SASL queue + background callbacks for OAUTHBEARER are wired correctly

Enabling:

• rd_kafka_conf_enable_sasl_queue(kafka_conf, 1) at config time, and
• rd_kafka_sasl_background_callbacks_enable(ctx->kafka.rk) after rd_kafka_new()

for all OAUTHBEARER uses (not just MSK IAM) is a solid way to ensure OAuth token refresh on idle connections or long poll intervals. Logging both the enable and failure cases is also helpful.

No issues from a lifecycle perspective: both calls are made before producing/consuming, and failures degrade gracefully to the previous behavior.

Please confirm your minimum supported librdkafka version exposes rd_kafka_conf_enable_sasl_queue() and rd_kafka_sasl_background_callbacks_enable(); otherwise, this will need version guards.

Also applies to: 412-424

---

358-389: MSK IAM registration is now correctly gated on explicit opt‑in and AWS brokers

The new registration block only activates MSK IAM when:

• ctx->aws_msk_iam is set (i.e., user explicitly configured rdkafka.sasl.mechanism=aws_msk_iam), and
• the effective mechanism is OAUTHBEARER, and
• brokers look like AWS MSK endpoints (.kafka. or .kafka-serverless. plus .amazonaws.com).

Passing the brokers string into flb_aws_msk_iam_register_oauth_cb() and cleaning up ctx->msk_iam in the init_error path closes the earlier resource-leak and “forced MSK IAM on generic OAUTHBEARER” issues. This is a good, minimal, and safe gating.

---

391-403: Correct handling of kafka_conf ownership and MSK IAM cleanup on init failure

Two important fixes here:

• kafka_conf is left non-NULL when rd_kafka_new() fails and only nulled on success. The init_error handler now:

  • Destroys the consumer (rd_kafka_destroy) when created, or
  • Calls rd_kafka_conf_destroy(kafka_conf) when creation failed.

• ctx->msk_iam is now destroyed in the init_error path (under FLB_HAVE_AWS_MSK_IAM), mirroring in_kafka_exit() and preventing leaks of TLS/aws-provider resources when init fails after successful MSK IAM setup.

This aligns with librdkafka’s ownership rules and fixes the reported leak.

Given this relies on librdkafka semantics (conf is only consumed on successful rd_kafka_new()), please verify against the version you build with.

Also applies to: 486-507

plugins/out_kafka/kafka_config.c (4)

197-208: Opaque usage and SASL queue enablement for OAUTHBEARER are correct

• ctx->opaque is initialized and wired into the conf; flb_kafka_opaque_set(ctx->opaque, ctx, NULL) keeps ptr pointing to the plugin context while leaving msk_iam_ctx free for MSK IAM use.
• Enabling the SASL queue for any ctx->sasl_mechanism == "OAUTHBEARER" ensures librdkafka’s OAuth token refresh machinery can run even when the producer is idle.

This matches the input plugin behavior and is consistent with the new MSK IAM OAuth callback flow.

Also applies to: 209-218

---

220-250: MSK IAM registration is correctly scoped and cleans up on failure

The producer side only registers MSK IAM when:

• The user explicitly requested it (ctx->aws_msk_iam), and
• The effective mechanism is OAUTHBEARER, and
• Brokers look like AWS MSK endpoints (.kafka. or .kafka-serverless. plus .amazonaws.com).

If flb_aws_msk_iam_register_oauth_cb() fails, calling flb_out_kafka_destroy(ctx) and returning NULL cleanly tears down all partially-initialized resources (conf, opaque, topics, etc.). The follow-up sasl.oauthbearer.config setting is also correctly guarded and failure is logged but non-fatal.

This resolves the earlier concern about silently enabling MSK IAM for generic OAUTHBEARER configs.

---

252-267: rd_kafka_conf ownership and destroy paths are now correct

The updated logic around rd_kafka_new() and teardown:

• Leaves ctx->conf intact when rd_kafka_new() fails so flb_out_kafka_destroy() can call rd_kafka_conf_destroy(ctx->conf).
• Sets ctx->conf = NULL only on success, relying on rd_kafka_destroy(ctx->kafka.rk) to free the configuration later.
• In flb_out_kafka_destroy(), destroys ctx->kafka.rk when present, otherwise destroys ctx->conf.

This matches librdkafka’s contract and closes the memory-leak scenario previously flagged in the producer path.

Please validate this against your librdkafka version’s rd_kafka_new() docs to ensure the “conf consumed only on success” assumption still holds.

Also applies to: 332-351

---

268-285: Background callbacks and MSK IAM teardown

Enabling rd_kafka_sasl_background_callbacks_enable() for OAUTHBEARER and checking its return value provides robust behavior:

• On success, tokens are refreshed even on idle producers.
• On failure, you emit a clear warning but keep the producer usable.

The MSK IAM context (ctx->msk_iam) is destroyed in flb_out_kafka_destroy(), after rd_kafka_destroy(), which ensures the OAuth callback will not be invoked again after the context is freed. This is a sensible ordering.

Because this depends on librdkafka background callback semantics, please confirm that rd_kafka_destroy() guarantees no further OAuth callbacks after it returns in your target version.

Also applies to: 363-369

src/aws/flb_aws_msk_iam.c (5)

137-212: Region extraction from broker covers standard, serverless, and VPC endpoint forms

extract_region_from_broker():

• Strips any :port suffix.
• Locates .amazonaws.com, with a special case for .vpce.amazonaws.com.
• Walks back to the previous . to isolate the region between that dot and .amazonaws.com (or .vpce for VPC endpoints).
• Handles:
  • Standard MSK: ...kafka.<region>.amazonaws.com
  • Serverless: ...kafka-serverless.<region>.amazonaws.com
  • VPC: ...kafka.<region>.vpce.amazonaws.com
  • Partitions where the host continues past amazonaws.com (e.g. amazonaws.com.cn), since the substring match is against .amazonaws.com.

Length checks and NULL returns on malformed inputs look sane, and error paths are clean. This is a reasonable heuristic for AWS MSK brokers and aligns with how plugins gate MSK IAM usage based on .kafka[‑serverless].*amazonaws.com.

---

215-512: SigV4-style payload builder is consistent and cleans up correctly

build_msk_iam_payload() now:

• Validates presence of config->region, host, and credentials.
• Constructs the credential string AKIA/yyyymmdd/region/kafka-cluster/aws4_request.
• Builds the canonical query (Action, X-Amz-Algorithm, Credential, Date, Expires, optional Security-Token, SignedHeaders=host).
• Computes:
  • Empty payload hash.
  • Canonical request over GET\n/\n<query>\nhost:<host>\n\nhost\n<hash>.
  • String to sign and signing key using the AWS standard AWS4 HMAC chain with service kafka-cluster.
  • Hex signature and appends X-Amz-Signature.
• Constructs the presigned URL and then Base64-encodes it, converting to URL-safe Base64 and trimming padding.

The error label frees every intermediate flb_sds_t, the success path also destroys all temporaries, and payload is the only returned SDS. No obvious leaks or double-frees, and the logic matches AWS SigV4 for query-signing MSK IAM URLs.

Because correctness here is tied to AWS MSK IAM’s SigV4 expectations, please validate this flow (service name kafka-cluster, query parameters, and signing steps) against the latest AWS MSK IAM documentation or reference implementation.

---

514-625: OAuth refresh callback: concurrency, provider usage, and token lifetime

The new oauthbearer_token_refresh_cb:

• Verifies the opaque context and region.
• Derives the MSK host (kafka-serverless.<region>.amazonaws.com vs kafka.<region>.amazonaws.com) from config->is_serverless.
• Uses a dedicated pthread_mutex_t lock to serialize access to the AWS provider:
  • Locks config->lock.
  • Calls provider_vtable->refresh() and get_credentials().
  • Unlocks before building the payload, minimizing contention.
• Builds the MSK IAM payload with the just-fetched credentials.
• Sets an OAuth token lifetime of now + MSK_IAM_TOKEN_LIFETIME_SECONDS (5 minutes) and delegates refresh cadence to librdkafka’s background callbacks.
• Destroys the temporary credentials and payload in all paths.

This addresses prior race conditions between librdkafka’s callback thread and other provider users, and the fixed 5‑minute lifetime aligns with common AWS MSK IAM practice.

Please confirm:

• That your configured AWS provider chain is safe to call from this background thread, and
• That librdkafka’s rd_kafka_oauthbearer_set_token() semantics (absolute ms since epoch) match how md_lifetime_ms is being calculated, in your target version.

---

627-761: Broker-driven registration with TLS-backed provider and mutex is sound

flb_aws_msk_iam_register_oauth_cb() now:

• Requires:
  • Non-NULL opaque, and
  • Non-empty brokers string.
• Extracts the first broker from the CSV list and determines:
  • Region via extract_region_from_broker().
  • Whether it’s serverless via .kafka-serverless. substring.
• Allocates struct flb_aws_msk_iam, stores config, region, is_serverless, and logs what it detected.
• Creates a TLS context (ctx->cred_tls) dedicated to credential fetching.
• Builds a standard AWS provider chain (flb_standard_chain_provider_create) for the extracted region, calls sync + init, then switches to async.
• Initializes ctx->lock to guard provider access from other threads.
• Stores ctx in opaque->msk_iam_ctx while leaving opaque->ptr untouched, and wires the OAuth refresh callback via rd_kafka_conf_set_oauthbearer_token_refresh_cb().

All error paths:

• Free first_broker and region_str as appropriate.
• Destroy TLS and provider when partially initialized.
• Avoid leaving a half-initialized context behind.

The resulting design is broker-driven (no cluster ARN), supports Standard and Serverless, and encapsulates TLS and provider lifecycles cleanly.

Given this depends on flb_standard_chain_provider_create() and flb_tls_create() semantics, please verify that:

• The chosen TLS settings are appropriate for IMDS/STS endpoints in your environment, and
• The provider chain you construct here is the same one you rely on elsewhere for AWS integrations.

---

763-785: Destroy path fully matches new allocations

flb_aws_msk_iam_destroy() now:

• Destroys ctx->provider.
• Destroys the TLS context ctx->cred_tls.
• Frees ctx->region.
• Destroys the mutex.
• Frees the struct itself.

This mirrors all allocations made in flb_aws_msk_iam_register_oauth_cb() and avoids leaks or dangling state.

Reviewing files that changed from the base of the PR and between 491287a and 776e6c5.

• .github/scripts/commit_prefix_check.py (4 hunks)
• .github/scripts/tests/test_commit_lint.py (4 hunks)
• .github/workflows/commit-lint.yaml (2 hunks)
• CMakeLists.txt (1 hunks)
• dockerfiles/Dockerfile (11 hunks)
• include/fluent-bit/aws/flb_aws_msk_iam.h (1 hunks)
• plugins/in_elasticsearch/in_elasticsearch.c (1 hunks)
• plugins/in_exec_wasi/in_exec_wasi.c (1 hunks)
• plugins/in_forward/fw.c (3 hunks)
• plugins/in_forward/fw_prot.c (2 hunks)
• plugins/in_kafka/in_kafka.c (4 hunks)
• plugins/in_kafka/in_kafka.h (1 hunks)
• plugins/in_node_exporter_metrics/ne_utils.c (1 hunks)
• plugins/out_kafka/kafka.c (0 hunks)
• plugins/out_kafka/kafka_config.c (4 hunks)
• plugins/out_kafka/kafka_config.h (1 hunks)
• src/aws/flb_aws_credentials_ec2.c (1 hunks)
• src/aws/flb_aws_credentials_sts.c (2 hunks)
• src/aws/flb_aws_msk_iam.c (12 hunks)
• src/flb_kafka.c (1 hunks)

• plugins/out_kafka/kafka.c

plugins/in_elasticsearch/in_elasticsearch.c (1)

240-240: Unrelated change: This file is outside the PR's stated scope.

This documentation improvement is correct and helpful, but the Elasticsearch input plugin appears unrelated to the PR's stated objective of implementing AWS MSK IAM authentication for Kafka plugins.

Was this change intentionally included, or should it be moved to a separate PR for documentation improvements?

.github/workflows/commit-lint.yaml (1)

16-16: ⚠️ Breaking changes in actions/setup-python@v6 require verification before merging.

The actions/setup-python@v6 upgrade introduces a breaking change: Node.js runtime upgrade from v20 to v24, which requires GitHub Actions runner v2.327.1 or later. The actions/checkout@v6 also requires runner v2.329.0+ for authenticated git operations.

Before deploying:

1. Verify your GitHub Actions runners meet the minimum version requirement (v2.329.0+)
2. Test this workflow on a branch with the appropriate runner version
3. Review and test any caching steps—there are known compatibility issues between actions/setup-python@v6 and some cache integrations; migrate to actions/cache@v4+ if needed

The commit prefix validation failures in PR comments are unrelated to these version upgrades; they reflect commit message format issues in the actual commits.

plugins/in_forward/fw.c (1)

484-519: Changes appear unrelated to PR scope but are technically correct.

The config map description improvements and the FLB_CONFIG_MAP_MULT flag for security.users are technically sound. However, this forward input plugin has no connection to Kafka or AWS MSK IAM authentication, which is the stated purpose of this PR. Consider whether these changes should be in a separate PR focused on forward plugin improvements.

plugins/in_forward/fw_prot.c (1)

1148-1153: Trace decode error handling and ownership transfer look correct

Switching the check to ret != CTR_DECODE_MSGPACK_SUCCESS brings the traces path in line with the metrics path and correctly treats any non‑success as a hard failure. Removing ctr_decode_msgpack_destroy(ctr) on the success path (while still destroying ctr on append failure) avoids double‑free and matches the documented ownership transfer to flb_input_trace_append().

One thing to confirm: that ctr_decode_msgpack_create() only initializes/allocates ctr on CTR_DECODE_MSGPACK_SUCCESS, so returning immediately on non‑success cannot leak anything. If that contract ever changes, we’d need to add a destroy in the error path as well.

Also applies to: 1158-1163

src/aws/flb_aws_credentials_ec2.c (1)

128-139: LGTM!

Minor formatting change (empty line addition) with no functional impact.

src/aws/flb_aws_credentials_sts.c (2)

173-186: LGTM!

Minor formatting change (empty line addition) with no functional impact on the STS provider refresh logic.

---

478-489: LGTM!

Minor formatting change (empty line addition) consistent with other provider implementations.

CMakeLists.txt (1)

86-93: Good security hardening addition.

Enabling --noexecstack by default on Linux prevents executable stack warnings and strengthens security posture. The guard correctly excludes macOS where these linker flags aren't supported.

Note: Similar flags exist in FLB_WASM_STACK_PROTECT (line 101) and FLB_SECURITY (line 502) sections, but having them unconditionally applied ensures baseline protection regardless of those optional settings.

.github/scripts/commit_prefix_check.py (2)

30-85: Well-structured refactor of prefix inference.

The separation of component_prefixes and build_seen tracking provides clear semantics:

• Component prefixes track actual code changes (lib, tests, plugins, src)
• build_optional correctly indicates when build: prefix can be omitted

The return type change from a single set to a tuple is a breaking change, but the corresponding test updates handle this appropriately.

---

64-73: Verify src/ subdirectory prefix inference behavior.

For src/ files not starting with flb_, the subdirectory name becomes the prefix (e.g., src/aws/file.c → aws:). This may produce unexpected prefixes for some subdirectories.

Consider whether all src/ subdirectories should generate prefixes this way, or if certain directories (like aws) should have special handling.

.github/scripts/tests/test_commit_lint.py (3)

37-39: LGTM!

Test correctly updated to unpack the new tuple return value and assert both prefixes and build_optional.

---

250-262: Test properly validates multi-component rejection.

The updated test and assertion correctly verify that commits touching multiple non-build components are rejected, with the error message checking for "does not match files changed".

---

290-313: Good edge case coverage for build_optional semantics.

The tests correctly verify:

• Empty file list → build_optional = False
• Unknown paths → build_optional = False
• Real component paths → build_optional = True

This ensures the build_optional flag behaves as documented.

src/flb_kafka.c (1)

96-99: Correct cleanup for rd_kafka_conf_t in error path

Switching from raw flb_free(kafka_cfg) to rd_kafka_conf_destroy(kafka_cfg) correctly matches the object’s allocation API and avoids undefined behavior.

plugins/in_kafka/in_kafka.h (1)

57-60: Explicit MSK IAM opt‑in flag looks correct

Adding aws_msk_iam as an explicit flag on the input config cleanly supports opt‑in semantics and matches the gating logic in in_kafka.c.

include/fluent-bit/aws/flb_aws_msk_iam.h (1)

37-49: Header/API now matches broker‑driven MSK IAM design

The updated flb_aws_msk_iam_register_oauth_cb signature and parameter docs (config, kconf, opaque, brokers) match the implementation and correctly expose the broker‑based activation flow.

plugins/out_kafka/kafka_config.h (1)

128-131: Output config mirrors input’s explicit MSK IAM flag

Introducing aws_msk_iam alongside msk_iam on the output side keeps semantics consistent with the input plugin and enables explicit IAM opt‑in.

dockerfiles/Dockerfile (1)

22-60: Debian 13/distroless 13 migration and jemalloc options wiring look coherent

The move to debian:trixie-slim/cc-debian13, updated lib* t64 package set, and the optional FLB_JEMALLOC_OPTIONS plumbed through to CMake form a consistent story for newer bases and allocator tuning. Please just confirm all listed packages (especially libssl3t64, libcurl4t64, libgnutls30t64, libldap2) exist on all target architectures for your build matrix.

Also applies to: 74-77, 95-99, 121-178, 185-215, 216-269

plugins/in_kafka/in_kafka.c (2)

271-297: SASL/OAUTHBEARER handling and background refresh are well integrated

Reading rdkafka.sasl.mechanism up front, translating aws_msk_iam to OAUTHBEARER while recording explicit opt‑in, and enabling both the SASL queue and background callbacks for all OAUTHBEARER cases is a solid way to fix idle‑connection expiry without impacting non‑IAM OAuth users.

Also applies to: 347-356, 404-424

---

486-507: Init‑error path now correctly cleans up MSK IAM context

Adding flb_aws_msk_iam_destroy(ctx->msk_iam) to the init error path aligns the failure cleanup with in_kafka_exit() and prevents leaking the MSK IAM TLS/provider resources when initialization fails mid‑way.

plugins/out_kafka/kafka_config.c (2)

61-87: Producer SASL/MSK IAM handling mirrors the input side correctly

The logic that detects rdkafka.sasl.mechanism=aws_msk_iam, records explicit opt‑in, rewrites the mechanism to OAUTHBEARER, and defaults the security protocol to SASL_SSL is consistent with the input plugin and with librdkafka expectations.

---

252-267: rd_kafka_conf ownership now handled correctly on producer creation success/failure

Conditionally nulling ctx->conf only after a successful rd_kafka_new() and destroying either the producer handle or the conf (but not both) in flb_out_kafka_destroy() aligns with librdkafka’s ownership semantics and removes the earlier leak on creation failure.

Also applies to: 338-351

src/aws/flb_aws_msk_iam.c (3)

137-212: Broker‑driven region detection and MSK IAM context setup look solid

The new extract_region_from_broker() helper and the reworked flb_aws_msk_iam_register_oauth_cb() correctly switch MSK IAM activation to a broker‑driven model, avoid opaque type confusion, and initialize TLS + the AWS credential provider with appropriate cleanup on every failure path. Region parsing for standard, serverless, and VPC endpoint hostnames looks reasonable and well‑guarded by validation and logging.

Also applies to: 627-761

---

214-512: MSK IAM payload construction and cleanup look correct

The refactored build_msk_iam_payload() validates inputs, derives the SigV4 presigned URL, produces a URL‑safe Base64 payload, and has a thorough single‑exit error path that frees all intermediate SDS allocations. The memory‑management and hashing/HMAC steps look consistent and leak‑free; any further tweaks (e.g., preferring flb_sds_len over strlen on SDS values) are purely stylistic.

---

514-625: OAuth refresh callback now safely synchronizes provider access and token lifetime

Wrapping the credential provider refresh/get_credentials calls with a mutex and using a fixed 5‑minute token lifetime (backed by librdkafka’s background refresh at ~80%) resolves the earlier concurrency and idle‑expiry issues. The destroy function correspondingly cleans up provider, TLS, region, and the mutex, so the MSK IAM context’s lifecycle is well‑defined.

Also applies to: 763-785

Reviewing files that changed from the base of the PR and between a6d39b1 and 776e6c5.

• .github/scripts/commit_prefix_check.py (4 hunks)
• .github/scripts/tests/test_commit_lint.py (4 hunks)
• .github/workflows/commit-lint.yaml (2 hunks)
• CMakeLists.txt (1 hunks)
• dockerfiles/Dockerfile (11 hunks)
• include/fluent-bit/aws/flb_aws_msk_iam.h (1 hunks)
• plugins/in_elasticsearch/in_elasticsearch.c (1 hunks)
• plugins/in_exec_wasi/in_exec_wasi.c (1 hunks)
• plugins/in_forward/fw.c (3 hunks)
• plugins/in_forward/fw_prot.c (2 hunks)
• plugins/in_kafka/in_kafka.c (4 hunks)
• plugins/in_kafka/in_kafka.h (1 hunks)
• plugins/in_node_exporter_metrics/ne_utils.c (1 hunks)
• plugins/out_kafka/kafka.c (0 hunks)
• plugins/out_kafka/kafka_config.c (4 hunks)
• plugins/out_kafka/kafka_config.h (1 hunks)
• src/aws/flb_aws_credentials_ec2.c (1 hunks)
• src/aws/flb_aws_credentials_profile.c (1 hunks)
• src/aws/flb_aws_credentials_sts.c (2 hunks)
• src/aws/flb_aws_msk_iam.c (12 hunks)
• src/flb_kafka.c (1 hunks)

• plugins/out_kafka/kafka.c

• include/fluent-bit/aws/flb_aws_msk_iam.h
• plugins/in_kafka/in_kafka.h

plugins/in_elasticsearch/in_elasticsearch.c (1)

237-241: Config help text update is clear and consistent

The added description for the http2 option is accurate and matches surrounding config_map style. No further changes needed.

src/aws/flb_aws_credentials_ec2.c (1)

132-134: LGTM!

This is a minor formatting change adding a blank line for readability. No functional impact.

src/aws/flb_aws_credentials_sts.c (1)

173-186: Whitespace-only change; no functional impact

The added blank lines in refresh_fn_sts and refresh_fn_eks are purely cosmetic and do not affect behavior, error handling, or locking. Change looks fine as-is.

Also applies to: 478-489

plugins/in_forward/fw_prot.c (2)

1149-1151: Unrelated changes bundled in this PR, but the improvements look good.

This file (plugins/in_forward/fw_prot.c) is unrelated to the PR's stated objectives of adding AWS MSK IAM authentication for the Kafka plugin. However, the changes themselves are improvements:

• Using CTR_DECODE_MSGPACK_SUCCESS constant instead of magic number -1 improves code clarity.
• Using flb_plg_error provides proper plugin-scoped logging.

Consider separating unrelated changes into a dedicated PR to maintain clear change history and simplify review.

---

1162-1162: Good documentation of ownership semantics.

The ownership comment clarifies that flb_input_trace_append takes ownership of the ctr object and destroys it on success, preventing a double-free bug. This is a valuable addition for maintainability.

Note that this differs from the METRICS path (lines 1141-1144), where cmt_decode_msgpack_destroy(cmt) is called explicitly after flb_input_metrics_append. This asymmetry appears intentional based on different ownership semantics between the two APIs.

As noted previously, these changes are unrelated to the PR's AWS MSK IAM objectives.

dockerfiles/Dockerfile (5)

74-77: Jemalloc configuration is properly wired.

The ARG → ENV → CMake propagation pattern is correct. The optional nature of FLB_JEMALLOC_OPTIONS is clearly documented in the comment. No issues identified.

Also applies to: 95-95

---

216-216: Debug stage updates align with trixie migration.

The debug stage properly includes t64-suffixed runtime libraries and appropriate development tools. The addition of LDAP support (libldap2) is consistent with other stages.

Also applies to: 237-268

---

185-185: Distroless image upgrade to cc-debian13 is confirmed and appropriate.

The gcr.io/distroless/cc-debian13 image is available and actively maintained by Google Container Tools with support across multiple architectures and variant tags (nonroot, debug). This aligns properly with the Debian trixie baseline.

---

22-22: Verify Debian trixie-slim compatibility before merging.

The upgrade from bookworm-slim to trixie-slim involves significant OS-level changes documented by Debian: glibc ABI bumps, package removals, and potential service configuration incompatibilities. Ensure:

1. All build dependencies (build-essential, libssl-dev, libcurl4-openssl-dev, libsystemd-dev, etc.) resolve correctly in trixie
2. The CMake toolchain is compatible with trixie's compiler versions
3. The distroless cc-debian13 image is available and properly tested
4. All package name updates with t64 suffixes are complete across builder, deb-extractor, and debug stages

---

129-170: Both libnghttp2-14 and libnghttp3-9 are intentionally included and do not require changes.

The deb-extractor stage correctly includes both libraries: libnghttp2 provides HTTP/2 support while libnghttp3 provides HTTP/3 support. This dual-library pattern is a standard practice in HTTP clients (such as curl) that need to support multiple protocol versions with automatic fallback. This is not an inconsistency.

The t64-suffixed packages (libssl3t64, libcurl4t64, etc.) are correctly named for the Debian trixie 64-bit time_t ABI transition and are the appropriate versions for trixie-slim base images. The LDAP packages (libldap-2.5 in the builder, libldap2 in deb-extractor) are properly included for MSK IAM SASL authentication support in the distroless multi-stage build pattern, where runtime shared libraries must be extracted and copied into the runtime image.

Likely an incorrect or invalid review comment.

plugins/in_node_exporter_metrics/ne_utils.c (1)

156-157: LGTM!

The buffer size increase from 512 to 2048 bytes is reasonable for handling longer lines in proc/sys filesystem files, and aligns with the existing real_path buffer size for consistency.

src/aws/flb_aws_credentials_profile.c (1)

664-673: LGTM!

Downgrading the log level to AWS_CREDS_DEBUG for a missing credentials file (ENOENT) is appropriate. This is an expected condition in environments using alternative credential sources (IAM roles, environment variables, ECS/EC2 metadata). The function still correctly returns -1 to signal the caller to try other credential providers.

CMakeLists.txt (1)

86-93: LGTM! Good security hardening.

Adding -Wa,--noexecstack and -Wl,-z,noexecstack flags unconditionally for Linux builds is a solid security improvement. This prevents executable stack vulnerabilities and silences kernel warnings. The platform check correctly excludes macOS where these flags aren't supported.

plugins/in_forward/fw.c (1)

484-519: LGTM!

The configuration description updates improve clarity by providing better context about secure forward authentication. These documentation enhancements will help users understand the purpose of each option.

src/flb_kafka.c (1)

96-100: Good fix for proper resource cleanup.

Using rd_kafka_conf_destroy() instead of flb_free() is correct. The rd_kafka_conf_t object is allocated by librdkafka via rd_kafka_conf_new() and contains internal structures that must be freed using the library's destructor to avoid memory leaks and corruption.

.github/scripts/tests/test_commit_lint.py (5)

37-39: LGTM!

Tests correctly updated to unpack the new tuple return value (prefixes, build_optional) from infer_prefix_from_paths(). The assertion for build_optional is True is correct since plugin files represent real components.

---

61-67: LGTM!

Test correctly verifies that multiple files from different components produce multiple prefixes, and that build_optional is True when real components are touched.

---

250-261: LGTM!

Test and assertion correctly updated. The error message check changed from "Expected one of:" to "does not match files changed" which aligns with the updated validate_commit logic that rejects multi-component commits requiring split.

---

294-297: LGTM!

Correctly verifies that an empty file list returns an empty prefix set with build_optional as False, since no components or CMakeLists.txt were touched.

---

607-622: LGTM!

Test docstring and logic correctly updated to reflect that multi-component commits must be split, even when the subject matches one component. The assertions properly verify both expected prefixes are reported.

.github/workflows/commit-lint.yaml (2)

26-26: No action needed. actions/setup-python@v6 is a valid version (currently at v6.1.0) and will not cause CI failures.

---

16-16: No action needed — actions/checkout@v6 is a valid, stable release.

As of December 2025, actions/[email protected] has been released and is stable. Similarly, actions/setup-python@v6 (v6.1.0 as of late November 2025) is also available. These version bumps are legitimate and will not cause workflow failures.

Likely an incorrect or invalid review comment.

plugins/out_kafka/kafka_config.h (1)

128-131: AWS MSK IAM fields are well-scoped and clear

Keeping both msk_iam and the aws_msk_iam flag under #ifdef FLB_HAVE_AWS_MSK_IAM cleanly scopes AWS‑specific state and the comment makes the explicit‑opt‑in semantics obvious. Looks good and consistent with the input plugin struct.

.github/scripts/commit_prefix_check.py (1)

160-213: Multi‑component and build: prefix rules are consistent with inferred prefixes

The new logic that:

• filters out build:/cmakelists.txt: from non_build_prefixes,
• requires an umbrella lib: prefix when multiple non‑build areas are touched, and
• enforces build: only when build_optional is false

gives clear, deterministic behavior for mixed commits while still allowing flexible subjects when only docs/tools are touched (expected empty). The error messaging listing Expected one of: ... will also be helpful when commits span several areas.

src/aws/flb_aws_msk_iam.c (5)

137-212: Region extraction from broker covers Standard, Serverless, and VPC endpoint formats

The extract_region_from_broker() logic that:

• strips the optional :port,
• locates .amazonaws.com (or .vpce.amazonaws.com via the .vpce check), and
• walks back to the preceding dot to slice out the region

correctly handles the documented MSK Standard, Serverless, and VPC endpoint hostnames while rejecting malformed or path‑like inputs. The explicit length sanity check on the region string is a nice guardrail.

---

215-512: MSK IAM payload construction and cleanup look correct; consider double‑checking expiry choices

The refactored build_msk_iam_payload():

• Validates config->region, host, and credentials up front.
• Builds the credential scope, canonical query (including optional session token), canonical request, and signing key following AWS SigV4 conventions.
• Constructs the full presigned URL, base64‑encodes it, then normalizes it to URL‑safe Base64 and strips padding.
• Has a comprehensive error: path that frees every intermediate flb_sds_t and avoids double frees by nulling empty_payload_hex after the happy‑path destroy.

The only behavioral point worth a second look is that the presigned URL uses X-Amz-Expires=900 (15 minutes) while the OAuth token lifetime is fixed at 300 seconds via MSK_IAM_TOKEN_LIFETIME_SECONDS. Running the OAuth lifetime shorter than the presign expiry is likely safe (and conservative), but it would be good to confirm this matches the latest AWS MSK IAM recommendations.

If you haven’t already, please verify against the current MSK IAM docs that a 5‑minute OAuth token lifetime combined with a 900‑second X-Amz-Expires is the intended combination (or adjust one of them if AWS expects them to match).

---

514-625: OAuth refresh callback now safely serializes provider access and cleans up credentials

The updated oauthbearer_token_refresh_cb() looks solid:

• It validates the opaque context and region and derives the MSK endpoint host based on is_serverless.
• It uses pthread_mutex_lock/unlock around provider->refresh() and get_credentials() so librdkafka’s background thread doesn’t race with any other potential provider users.
• It bails early with rd_kafka_oauthbearer_set_token_failure() and clear log messages when refresh or credentials retrieval fails.
• It hands credentials to build_msk_iam_payload(), sets a token lifetime based on MSK_IAM_TOKEN_LIFETIME_SECONDS as an absolute epoch time, and always destroys the credentials and payload afterward.

This is a good concurrency and lifecycle story for the MSK IAM provider.

Since this callback is driven by librdkafka internals, please ensure your shutdown paths (rd_kafka_destroy for producer/consumer followed by flb_aws_msk_iam_destroy) are exercised in tests so there’s no window where the provider is freed while a background refresh callback might still be in flight.

---

627-761: MSK IAM registration lifecycle and opaque usage are clean; just confirm flb_kafka_opaque_set semantics

The registration function now:

• Validates opaque and brokers explicitly.
• Parses the first broker, extracts the region via extract_region_from_broker(), and detects Serverless vs Standard via .kafka-serverless..
• Allocates ctx, sets ctx->region, creates a TLS instance, and wires a standard AWS provider (sync/init/async), aborting with full cleanup on each failure path.
• Initializes a mutex guarding provider access.
• Calls flb_kafka_opaque_set(opaque, NULL, ctx) and then registers both the opaque and the OAuth refresh callback on the conf.
• Provides a matching flb_aws_msk_iam_destroy() that tears down provider, TLS, region, mutex, and the struct itself.

The only thing to double‑check is how flb_kafka_opaque_set() behaves when called twice: Kafka plugins first call flb_kafka_opaque_set(ctx->opaque, ctx, NULL) to store the plugin context, and here you call it again with (opaque, NULL, ctx) to add the MSK IAM context. As long as flb_kafka_opaque_set() only overwrites the non‑NULL argument and leaves the existing plugin pointer intact, this is perfect; if it instead blindly overwrites both fields, you’d lose the plugin context for log/message callbacks.

It’s worth quickly inspecting flb_kafka_opaque_set() (in src/flb_kafka.c) to confirm it treats NULL arguments as “don’t change this field”; if it does not, updating it to do so will prevent clobbering opaque->ptr when MSK IAM is registered.

---

763-785: MSK IAM destroy path matches initialization and avoids leaks

flb_aws_msk_iam_destroy() mirrors the successful registration path by:

• Destroying the AWS provider if present,
• Destroying the TLS context if present,
• Freeing the region SDS,
• Destroying the mutex, and
• Finally freeing the struct.

Given that the registration function only returns a fully initialized ctx, this cleanup order is appropriate and should prevent leaks of TLS/provider resources.